#pragma once
#include <vector>
#include <cassert>
#include <string>
#include <cstring>
#include <iostream>
#include <ctime>
#include <functional>
#include <unordered_map>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <cstdio>
#include <stdint.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <signal.h>

#define INFO 0
#define DEBUG 1
#define ERROR 2
#define PRINT_LOG_LEVEL ERROR
// char thread_name[64];
// snprintf(thread_name, sizeof(thread_name), "/home/wxd/High-Concurrency-Server/Log/%p.log", (void *)pthread_self());
// FILE* fp = fopen(thread_name,"a");
// fclose(fp);

//"[%s %d] " format  format前的空格不能省略
#define LOG(loglevel, format, ...)                                                                                          \
    do                                                                                                                      \
    {                                                                                                                       \
        if (loglevel < PRINT_LOG_LEVEL)                                                                                     \
            break;                                                                                                          \
        time_t t = time(NULL);                                                                                              \
        struct tm *ltm = localtime(&t); /*时间戳转结构*/                                                                    \
        char str_time[32];                                                                                                  \
        strftime(str_time, sizeof(str_time) - 1, "%H:%M:%S", ltm); /*结构转字符串*/                                         \
        fprintf(stdout, "[%p %s %s %d] " format "\n", (void *)pthread_self(), str_time, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INFO_LOG(format, ...) LOG(INFO, format, ##__VA_ARGS__);
#define DEBUG_LOG(format, ...) LOG(DEBUG, format, ##__VA_ARGS__);
#define ERROR_LOG(format, ...) LOG(ERROR, format, ##__VA_ARGS__);

class Any
{
private:
    // 内部的父类
    class Holder
    {
    public:
        virtual ~Holder() {};
        // 构造新的子类返回（用于Any的拷贝构造）
        virtual Holder *CopyNewOne() = 0;
        // 返回子类的类型-注意type_info拷贝构造被删除，用引用接受（用于对比Get时的类型是否正确）
        virtual const std::type_info &GetTypeInfo() = 0;
    };

    // 子类-未来会存任意的类型
    template <class T>
    class PalceHolder : public Holder
    {
    public:
        // 直接根据对象构造
        PalceHolder(const T &val) : _val(val)
        {
        }
        ~PalceHolder() {};
        // 构造新的子类返回（用于Any的拷贝构造）
        virtual Holder *CopyNewOne() override
        {
            return new PalceHolder(_val);
        }

        virtual const std::type_info &GetTypeInfo() override
        {
            return typeid(T);
        }

    public:
        T _val; // 实际存储的任意T类型
    };

    void Swap(Any &other)
    {
        std::swap(_content, other._content);
    }

public:
    // 空构造函数
    Any() : _content(nullptr)
    {
    }
    // 根据任意对象直接构造
    template <class T>
    Any(const T &val) : _content(new PalceHolder<T>(val))
    {
    }
    // 拷贝构造
    Any(const Any &other) : _content(other._content ? other._content->CopyNewOne() : nullptr) // 如果是空Any对象，则直接赋值空，如果不是，则根据子类构造一个新的子类返回
    {
    }
    ~Any()
    {
        if (_content)
            delete _content;
    }
    // 赋值操作符-根据对象直接赋值
    template <class T>
    Any &operator=(const T &val)
    {
        Any any(val);
        Swap(any); // 直接根据val构造一个临时对象与之交换，临时对象会自动销毁
        return *this;
    }
    //  赋值操作符-根据Any对象赋值
    Any &operator=(Any other)
    {
        // 直接传值接收，会先调用拷贝构造，构造出临时的对象other，再交换即可，省去主动delete的过程
        Swap(other);
        return *this;
    }

    // 获取存储的对象，外部会显式传递类型
    template <class T>
    T *Get()
    {
        // 断言一下外部传的类型一定是与存储的类型相同
        assert(typeid(T) == _content->GetTypeInfo());
        if (_content == nullptr)
            return nullptr;
        // 返回内部的对象
        return &(((PalceHolder<T> *)_content)->_val);
    }

    Holder *_content;
};

#define BUFFER_INIT_SIZE 1024

class Buffer
{
private:
    // 获取起始位置的地址
    char *BeginPtr()
    {
        return &*_buffer.begin();
    }

    // 获取写位置的地址
    char *WritePtr()
    {
        return BeginPtr() + _w_index;
    }

    // 获取前部可写空间大小
    uint64_t FrontLdleSize()
    {
        return _r_index;
    }
    // 获取后部可写空间大小
    uint64_t BackLdleSize()
    {
        return _buffer.size() - _w_index;
    }
    // 获取剩余可写空间大小
    uint64_t LdleWriteSize()
    {
        return FrontLdleSize() + BackLdleSize();
    }
    // 移动写偏移
    void UpdateWriteOffset(uint64_t len)
    {
        assert(len <= BackLdleSize());
        _w_index += len;
    }

    // 对缓冲区预留指定写入空间 （移动 or 扩容）
    void Reserve(uint64_t len)
    {
        if (len <= BackLdleSize())
            return; // 后面的空间足够写入，不用预留
        if (len <= LdleWriteSize())
        {
            // 前后空间之和够用，把数据前移
            uint64_t od_size = ReadSize();
            std::copy(ReadPtr(), WritePtr(), BeginPtr());
            // 更新偏移量
            _r_index = 0;
            _w_index = _r_index + od_size;
        }
        else
        {
            // 空间完全不够用，直接扩容至足够写入，不用移动数据
            _buffer.resize(_w_index + len);
            DEBUG_LOG("resize:%ld", _w_index + len);
        }
    }

    // 向缓冲区写入
    void _Write(const void *data, uint64_t len)
    {
        // 先预留足够的空间
        if (len == 0)
            return;
        Reserve(len);
        // 拷贝进入缓冲区
        // 确保data每次++移动1字节
        const char *data_c = static_cast<const char *>(data);
        std::copy(data_c, data_c + len, WritePtr());
    }

    // 从缓冲区读取
    void _Read(void *buffer, uint64_t len)
    {
        // 确保读取的字节数不能超限
        assert(len <= ReadSize());
        // 拷贝出去
        std::copy(ReadPtr(), ReadPtr() + len, (char *)buffer);
    }

    char *FindSepPtr(char sep)
    {
        // void *memchr(const void *s, int c, size_t n);
        char *sep_ptr = (char *)memchr(ReadPtr(), sep, ReadSize());
        return sep_ptr;
    }

public:
    Buffer() : _buffer(BUFFER_INIT_SIZE), _r_index(0), _w_index(0)
    {
    }

    // 获取读位置的地址 (未来把发送缓冲区的数据发到socket缓冲区中需要用到)
    char *ReadPtr()
    {
        return BeginPtr() + _r_index;
    }
    // 移动读偏移 (未来把发送缓冲区的数据发到socket缓冲区中需要用到)
    void UpdateReadOffset(uint64_t len)
    {
        if (len == 0)
            return;
        assert(len <= ReadSize());
        _r_index += len;
    }
    // 缓冲区内实际存储数据数据的大小
    uint64_t ReadSize()
    {
        return _w_index - _r_index;
    }
    // 输入任意类型数据
    void Write(const void *data, uint64_t len)
    {
        if (len == 0)
            return;
        // 写入数据
        _Write(data, len);
        // 更新偏移
        UpdateWriteOffset(len);
    }
    // 提供输入字符串的接口（方便操作）
    void Write(const std::string &input_str)
    {
        _Write(input_str.c_str(), input_str.size());
        UpdateWriteOffset(input_str.size());
    }
    // 提供直接传入Buffer对象的接口
    void Write(Buffer &buffer)
    {
        _Write(buffer.ReadPtr(), buffer.ReadSize());
        UpdateWriteOffset(buffer.ReadSize());
    }

    // 读出任意类型数据
    void Read(void *buffer, uint64_t len)
    {
        // 读出数据
        _Read(buffer, len);
        // 更新偏移
        UpdateReadOffset(len);
    }

    // 读出字符串
    std::string Read(uint64_t len)
    {
        std::string read_str;
        read_str.resize(len);
        _Read(&read_str[0], len);
        UpdateReadOffset(len);
        return read_str;
    }

    // 读出一行字符串（读取协议使用）
    std::string GetLine()
    {
        char *sep_ptr = FindSepPtr('\n');
        if (sep_ptr == nullptr)
        {
            // 没找到一行，直接返回空串不做处理，这个很重要，确保未来接收的都是完整的一行
            return "";
        }
        // 计算长度
        uint64_t len = sep_ptr - ReadPtr() + 1;
        // 取出一整行，包括换行符
        std::string read_str;
        read_str.resize(len);
        _Read(&read_str[0], len);
        // 更新偏移
        UpdateReadOffset(len);

        return read_str;
    }

private:
    std::vector<char> _buffer; // 存储数据的容器
    uint64_t _r_index;         // 读偏移
    uint64_t _w_index;         // 写偏移
};

#define DEF_SOCKFD -1
#define DEF_BACKLOG 1024
#define SOCKFD_NONBLOCK 1
// TcpSocket的封装（服务端 与 客户端 使用）
class TcpSocket
{
private:
    bool Create()
    {
        //_sockfd = socket(AF_INET,SOCK_STREAM,0);
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ERROR_LOG("TcpSocket Create Error!");
            return false;
        }
        // DEBUG_LOG("TcpSocket Create Success!");
        return true;
    }
    bool Bind(uint16_t port)
    {
        // 构建地址结构
        struct sockaddr_in addr_in;
        addr_in.sin_family = AF_INET;
        addr_in.sin_port = htons(port);
        // addr_in.sin_addr.s_addr = INADDR_ANY;//其实直接设置就行，毕竟都是0
        inet_pton(AF_INET, "0.0.0.0", &addr_in.sin_addr.s_addr);
        socklen_t addr_in_len = sizeof(addr_in);
        // 绑定
        int ret = bind(_sockfd, (sockaddr *)&addr_in, addr_in_len);
        if (ret < 0)
        {
            ERROR_LOG("TcpSocket Bind Error!");
            return false;
        }
        DEBUG_LOG("TcpSocket Bind Success!");
        return true;
    }
    bool Listen()
    {
        int ret = listen(_sockfd, DEF_BACKLOG);
        if (ret < 0)
        {
            ERROR_LOG("TcpSocket Listen Error!");
            return false;
        }
        DEBUG_LOG("TcpSocket Listen Success!");
        return true;
    }

    // 设置文件描述符非阻塞
    void SetNonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
    // 设置地址端口可重复使用
    void AddrReuse()
    {
        int opt = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &opt, sizeof(opt));
    }

public:
    TcpSocket() : _sockfd(DEF_SOCKFD)
    {
    }
    // 根据特定fd构造TcpSocket对象
    TcpSocket(int fd) : _sockfd(fd)
    {
    }
    // 服务端接收连接,不关注客户端信息
    int Accept()
    {
        int ret = accept(_sockfd, nullptr, nullptr);
        if (ret < 0)
        {
            ERROR_LOG("TcpSocket Accept Error!");
            return -1;
        }
        return ret;
    }
    // 客户端发起连接
    bool Connect(std::string ip_str, uint16_t port)
    {
        // 构建地址结构
        struct sockaddr_in addr_in;
        addr_in.sin_family = AF_INET;
        addr_in.sin_port = htons(port);
        inet_pton(AF_INET, ip_str.c_str(), &addr_in.sin_addr.s_addr);
        socklen_t addr_in_len = sizeof(addr_in);
        int ret = connect(_sockfd, (sockaddr *)&addr_in, addr_in_len);
        if (ret < 0)
        {
            ERROR_LOG("TcpSocket Connect Error!");
            return false;
        }
        DEBUG_LOG("Client Connect Success");
        return true;
    }

    // 如果文件描述符被设置为非阻塞模式（通过 fcntl 设置 O_NONBLOCK 标志），
    // 那么无论 recv 的 flags 参数如何设置，recv 函数本身都会以非阻塞方式运行。
    ssize_t Recv(void *buffer, size_t len)
    {
        ssize_t ret = recv(_sockfd, buffer, len, 0);

        if (ret <= 0)
        {
            // 读完了or被打断，不算错误在内部我们直接返回一个0，不返回-1了，使用的时候收到0就不做处理，收到-1则确认出问题了
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            // 异常情况处理
            ERROR_LOG("TcpSocket Recv Error!");
            return -1; // 按理来说ret==0表示对端断开连接，但是我们这里返回-1统一处理
        }
        else
        {
            // 读取正常，返回实际读取的字节数
            return ret;
        }
        return ret;
    }
    // 同样_sockfd已经设置了非阻塞，send可以不用再设置flag
    ssize_t Send(const void *buffer, size_t len)
    {
        int ret = send(_sockfd, buffer, len, 0);
        if (ret < 0)
        {
            // 缓冲区满了 or 被打断
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0; // 可以原谅，返回0
            }
            // 异常情况处理
            ERROR_LOG("TcpSocket Send Error!");
            return -1;
        }
        else
        {
            // 读取正常，返回实际写入的字节数
            return ret;
        }
        return ret;
    }

    // 用于服务端创建套接字
    int ServerCreate(uint16_t port, bool flag /*是否非阻塞*/)
    {
        if (Create() < 0)
            return false;
        if (flag == 1)
        {
            // 设置非阻塞
            SetNonBlock();
        }
        // 设置可重复利用地址端口
        AddrReuse();
        if (Bind(port) < 0)
            return false; // 端口构造时指定了，IP默认0
        if (Listen() < 0)
            return false;

        return this->Fd();
    }
    // 用于客户端创建套接字 同时连接服务端
    bool ClientCreate(std::string ip_str, uint16_t port)
    {
        if (Create() < 0)
            return false;
        // 客户端不需要绑定
        if (Connect(ip_str, port) < 0)
            return false;

        return true;
    }

    int Fd()
    {
        return _sockfd;
    }
    void Close()
    {
        if (_sockfd < 0)
            return;
        close(_sockfd);
    }

private:
    int _sockfd;
};

class Epoller;
class EventLoop; // EventLoop对连接操作
#define EPOLL_IGN_ANY 0
class Channel
{
    using callback_t = std::function<void()>;

private:
    // 这里要用到Epoller的成员函数，由于定义在后面，声明一下，具体实现在类外
    // 对Epoll模型的操作（写入内核）
    void UpdateToEpoller();
    void DeleteFromEpoller();

public:
    Channel(int fd, EventLoop *event_loop)
        : _fd(fd), _event_loop(event_loop), _revents({0}), _events({0})
    {
    }
    // 判断是否设置了读or写
    bool IsSetReadCare()
    {
        return (_events.events & EPOLLIN);
    }
    bool IsSetWriteCare()
    {
        return (_events.events & EPOLLOUT);
    }
    // 设置对读事件的关心

    void SetReadCare()
    {
        _events.data.fd = _fd;
        _events.events |= EPOLLIN;
        // 未来设置到EpollLoop的Epoller中
        UpdateToEpoller();
    }
    // 设置对写事件的关心
    void SetWriteCare()
    {
        _events.data.fd = _fd;
        _events.events |= EPOLLOUT;
        // 未来设置到EpollLoop的Epoller中
        UpdateToEpoller();
    }
    // 取消对读事件的关心
    void CancelReadCare()
    {
        _events.data.fd = _fd;
        _events.events &= (~EPOLLIN);
        UpdateToEpoller();
    }
    // 取消对写事件的关心
    void CancelWirteCare()
    {
        _events.data.fd = _fd;
        _events.events &= (~EPOLLOUT);
        UpdateToEpoller();
    }
    // 取消对全部事件的关心
    void CancelAllCare()
    {
        _events.events = EPOLL_IGN_ANY;
        UpdateToEpoller();
    }
    // 移除对文件描述符的关心（从红黑树中拿走）
    void DeleteFd()
    {
        DeleteFromEpoller();
    }

    void SetReadCallback(const callback_t &read_call)
    {
        _read_callback = read_call;
    }
    void SetWriteCallback(const callback_t &write_call)
    {
        _write_callback = write_call;
    }
    void SetErrorCallback(const callback_t &error_call)
    {
        _error_callback = error_call;
    }
    void SetCloseCallback(const callback_t &close_call)
    {
        _close_callback = close_call;
    }
    void SetAnyCallback(const callback_t &any_call)
    {
        _any_callback = any_call;
    }
    // // 设置读写就绪
    // void SetReadReady()
    // {
    //     _revents.events |= EPOLLIN;
    // }
    // void SetWriteReady()
    // {
    //     _revents.events |= EPOLLOUT;
    // }
    // 设置就绪事件，用于Epoller设置
    void SetRevents(epoll_event events)
    {
        _revents = events;
    }
    // 返回关注的事件（用于Epoller根据Channel添加）
    epoll_event *GetEvents()
    {
        return &_events;
    }
    // 事件处理-未来由EventLoop触发事件后直接调用
    void EventHandler()
    {
        // 未来读事件的处理不涉及到关闭连接的问题，要确保把没写完的数据发回去，或者写出错再关闭连接
        if ((EPOLLIN & _revents.events) || (EPOLLRDHUP & _revents.events) || (EPOLLPRI & _revents.events))
        {
            // DEBUG_LOG("%d,读事件触发",_fd);
            if (_read_callback)
                _read_callback();
        }
        // 可能会关闭连接的回调，调用一次即可
        if (EPOLLOUT & _revents.events)
        {
            DEBUG_LOG("%d,写事件触发", _fd);
            if (_write_callback)
                _write_callback();
        }
        else if (EPOLLHUP & _revents.events)
        {
            if (_close_callback)
                _close_callback();
        }
        else if (EPOLLERR & _revents.events)
        {
            if (_error_callback)
                _error_callback();
        }

        // 任意事件均调用(不用担心Write或Read等会对连接进行释放，因为所有的释放操作都改成了压入任务池最后执行)
        if (_any_callback)
            _any_callback();
    }
    int Fd()
    {
        return _fd;
    }

private:
    int _fd;
    struct epoll_event _events = {0};  // 记录关心的事件   //可以考虑只保存结构里面的32位的events,也可以考虑使用epoll_event，注意设置的时候设置epoll_event.fd即可。
    struct epoll_event _revents = {0}; // 接收的触发的事件
    EventLoop *_event_loop;            // 记录未来操作它的EventLoop对象(未来添加事件到其管理的Epoll模型，同时EventLoop也要管理连接，构造Channel)
    callback_t _read_callback = nullptr;
    callback_t _write_callback = nullptr;
    callback_t _close_callback = nullptr;
    callback_t _error_callback = nullptr;
    callback_t _any_callback = nullptr;
};

#define EPOLL_CREATE_NUM 1 // 无实际意义，大于0即可
#define EPOLL_MAX_EVENTS 1024
class Epoller
{
private:
    // 操作epoll
    void EpollCtl(int op, int fd, epoll_event *events)
    {
        int ret = epoll_ctl(_epfd, op, fd, events);
        if (ret < 0)
        {
            ERROR_LOG("Epoller epoll_ctl Error!");
            return;
        }
    }
    // 判断是否已经添加过了（用于判断这次是添加还是修改）
    bool ChannelExist(Channel *channel)
    {
        auto it = _all_channels.find(channel->Fd());
        if (it == _all_channels.end())
            return false;
        return true;
    }

public:
    // 构造创建Epoll模型
    Epoller()
    {
        _epfd = epoll_create(EPOLL_CREATE_NUM);
        if (_epfd < 0)
        {
            ERROR_LOG("Epoller epoll_create Error:%s", strerror(errno));
            exit(1); // 此问题理论不会出现，如果出现，强制退出进程
        }
    }
    // 添加 或 修改
    void Update(Channel *channel)
    {
        // 如果不存在添加 ，存在则修改
        if (ChannelExist(channel))
        {
            // 修改
            EpollCtl(EPOLL_CTL_MOD, channel->Fd(), channel->GetEvents());
        }
        else
        {
            // 添加到hash
            _all_channels.emplace(std::make_pair(channel->Fd(), channel));
            // 添加到内核
            EpollCtl(EPOLL_CTL_ADD, channel->Fd(), channel->GetEvents());
        }
    }
    void Delete(Channel *channel)
    {
        if (ChannelExist(channel))
        {
            // 从hash中删除
            _all_channels.erase(channel->Fd());
            // 删除
            EpollCtl(EPOLL_CTL_DEL, channel->Fd(), nullptr);
        }
    }

    // 用于EpollLoop调用，返回就绪的Channel队列
    void WaitRun(std::vector<Channel *> &ready_channels)
    {
        // DEBUG_LOG("开始监控");
        int fds = epoll_wait(_epfd, _ready_events_fds, EPOLL_MAX_EVENTS, -1 /*阻塞*/);
        if (fds < 0)
        {
            DEBUG_LOG("被信号打断");
            // 被信号打断的阻塞，重新等待即可
            if (errno == EINTR)
                return;
        }
        else if (fds == 0)
        {
            DEBUG_LOG("Timeout");
        }
        else
        {
            // DEBUG_LOG("有事件就绪");
            // DEBUG_LOG("fds = %d",fds);
            for (int i = 0; i < fds; i++)
            {
                // 把就绪的事件添加进入Channel，并返回就绪Channel队列给调用方
                auto it = _all_channels.find(_ready_events_fds[i].data.fd);
                assert(it != _all_channels.end());
                it->second->SetRevents(_ready_events_fds[i]);
                ready_channels.push_back(it->second); // 返回给调用方
                // 清空就绪队列
            }
        }
    }

private:
    int _epfd;                                              // epoll模型的fd
    struct epoll_event _ready_events_fds[EPOLL_MAX_EVENTS]; // 存储实际就绪的文件描述符的事件的就绪队列，提前开辟
    std::unordered_map<int, Channel *> _all_channels;       // 用作模块之间建立联系，记录所管理的Channel，一方面用于判断是否被添加，另一方面事件触发后需要找到对应的Channel
};

#define TIME_WHEEL_CAPACITI 60
class TimeTask;
using task_fun_t = std::function<void()>;
using TimeTaskPtr = std::shared_ptr<TimeTask>;
using TimeTaskWeakPtr = std::weak_ptr<TimeTask>;
using wheel_func_t = std::function<void()>;

// 供TimeWheel使用的类
class TimeTask
{
public:
    TimeTask(uint64_t id, uint32_t timeout, task_fun_t task_func, wheel_func_t wheel_delete_func)
        : _id(id), _timeout(timeout), _task_func(task_func), _wheel_delete_func(wheel_delete_func), _is_cancel(false)
    {
    }
    ~TimeTask()
    {

        if (_is_cancel == false)
            _task_func();     // 对象销毁时自动执行任务
        _wheel_delete_func(); // 对象销毁时也要把自己从哈希表中删除
    }
    uint64_t Id() { return _id; };
    uint32_t TimeOut() { return _timeout; };
    void Cancel()
    {
        _is_cancel = true;
    };

private:
    uint64_t _id;                    // 标识超时任务的唯一性
    uint32_t _timeout;               // 记录任务的超时时间
    task_fun_t _task_func;           // 记录执行的具体任务,由外部绑定传入
    wheel_func_t _wheel_delete_func; // 从时间轮的哈希表中删除，由TimeWheel传入
    bool _is_cancel;
};

// 提供给外部的类
class TimeWheel
{
private:
    int TimerfdReadHandler()
    {
        // 有可能其它文件描述符处理花费时间较长，导致处理到定时器文件描述符是，已经超时了很多次了（本来应该一秒就触发一次的）
        // read读取的是从上一次read之后超时的次数
        //  把内容清空
        uint64_t times;
        int ret = read(_timerfd, &times, 8);
        if (ret < 0)
        {
            ERROR_LOG("TimerfdReadHandler Error!");
            abort();
        }

        return times;
    }
    void OnTime()
    {
        // 根据实际超时的次数执行
        int times = TimerfdReadHandler();
        for (int i = 0; i < times; i++)
        {
            RunTimeTask();
        }
    }
    void StartTimeFd()
    {
        // 让timefd每一秒触发一次事件
        //  int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
        timespec it_value;
        it_value.tv_sec = 1;
        it_value.tv_nsec = 0;

        timespec it_interval;
        it_interval.tv_sec = 1;
        it_interval.tv_nsec = 0;

        itimerspec time;
        time.it_value = it_value;
        time.it_interval = it_interval;

        int ret = timerfd_settime(_timerfd, 0, &time, nullptr);
        if (ret < 0)
        {
            // ERROR_LOG("StartTimeFd Error!");
            std::cerr << "[" << __LINE__ << ":" << __FILE__ << "]StartTimeFd Error!" << std::endl;
            abort();
        }
    }

    void DeleteTimetask(uint64_t id)
    {
        // DEBUG_LOG("id = %ld DeleteTimetask", id);
        auto it = _all_timetask.find(id);
        if (it != _all_timetask.end())
        {
            _all_timetask.erase(it);
        }
    }

    void AddTimeTask(uint64_t id, uint32_t timeout, task_fun_t task_func)
    {
        // TimeTaskPtr _new_timetask = std::shared_ptr<TimeTask>(new TimeTask(id,timeout,task_func));
        // 根据参数创建任务对象，设置未来从_all_timetask删除自己的回调
        TimeTaskPtr _new_timetask_ptr = std::make_shared<TimeTask>(id, timeout, task_func, std::bind(&TimeWheel::DeleteTimetask, this, id));

        // 存入_all_timetask
        TimeTaskWeakPtr _timetask_weak_ptr = _new_timetask_ptr;
        _all_timetask.emplace(std::make_pair(_new_timetask_ptr->Id(), _timetask_weak_ptr));
        std::shared_ptr<TimeTask> ptr = _timetask_weak_ptr.lock();
        // DEBUG_LOG("id = %ld,add task ptr:%p", id, ptr.get());
        //  把任务放入_timewheel
        uint32_t pos = (_tick + _new_timetask_ptr->TimeOut()) % _capacity;
        _timewheel[pos].push_back(_new_timetask_ptr);
    }
    void RefreshTimeTask(uint64_t id) // 外部提供id，自动刷新对应id的任务
    {
        auto it = _all_timetask.find(id);
        if (it == _all_timetask.end())
        {
            return; // 没找到任务
        }
        std::shared_ptr<TimeTask> ptr = it->second.lock();
        // DEBUG_LOG("id = %ld,Refresh task ptr:%p", id, ptr.get());
        //  刷新任务，拿到它的weak_ptr -> shared_ptr
        TimeTaskPtr timetask_ptr = it->second.lock(); // weak_ptr -> shared_ptr
        // 拿到后插入_timewheel
        uint32_t pos = (_tick + timetask_ptr->TimeOut()) % _capacity;
        if (timetask_ptr)
            _timewheel[pos].push_back(timetask_ptr);
    }

    void CancelTimeTask(uint64_t id)
    {
        auto it = _all_timetask.find(id);
        if (it == _all_timetask.end())
        {
            return; // 没找到任务
        }
        TimeTaskPtr task_ptr = it->second.lock();
        // ERROR_LOG("id = %ld cancel search task_ptr :%p", id, task_ptr.get());
        if (task_ptr) // 一定要不为空再访问这个指针，如果TimeTask的析构的顺序是先执行任务，后DeleteTimetask，执行任务的时候可能会再次查找哈希表，因为最后一个shared_ptr此时被释放，则这个task_ptr取出来的时候是空指针
            task_ptr->Cancel();
    }

public:
    TimeWheel(EventLoop *event_loop /*时间轮关联的event_loop*/)
        : _capacity(TIME_WHEEL_CAPACITI), _tick(0), _timewheel(_capacity), _event_loop(event_loop)
    {
        // 初始化_timerfd
        _timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (_timerfd < 0)
        {
            ERROR_LOG("timerfd_create error！");
            abort();
        }
        // DEBUG_LOG("timerfd = %d",_timerfd);
        //  初始化_timerfd_channel
        _timerfd_channel = std::make_unique<Channel>(_timerfd, _event_loop);

        // 设置_timerfd_channel的读事件触发回调（数据全部读出来，并再次调用RunTimeTask）
        _timerfd_channel->SetReadCallback(std::bind(&TimeWheel::OnTime, this));
        // 启动读事件关心
        _timerfd_channel->SetReadCare();
        // 启动timerfd
        StartTimeFd();
    }

    // 定时器中有个_timers成员,定时器信息的操作有可能在多线程中进行,因此需要考虑线程安全问题
    // 不想加锁的话,那就把对定期的所有操作,都放到一个线程中进行
    void DeleteTaskInLoop(uint64_t id);
    void AddTaskInLoop(uint64_t id, uint32_t timeout, task_fun_t task_fun);
    void RefreshTaskInLoop(uint64_t id);
    void CancelTimeTaskInLoop(uint64_t id);
    // 注意这个函数存在线程安全问题，确保其由EventLoop所在线程执行
    bool TimeTaskExist(uint64_t id)
    {
        auto it = _all_timetask.find(id);
        if (it != _all_timetask.end())
        {
            std::shared_ptr<TimeTask> ptr = it->second.lock();
            // DEBUG_LOG("id = %ld search task ptr:%p", id, ptr.get());
            return true;
        }

        return false;
    }

    // 每秒执行一次
    void RunTimeTask()
    {
        // 执行数组中所有任务
        _timewheel[_tick].clear();
        // 移动秒针
        _tick = (_tick + 1) % _capacity;
    }

private:
    uint32_t _capacity;                                          // 记录时间轮容量
    std::vector<std::vector<TimeTaskPtr>> _timewheel;            // 记录时间轮数组
    std::unordered_map<uint64_t, TimeTaskWeakPtr> _all_timetask; // 记录当前所有的任务对象（用于未来刷新Ptr）
    uint32_t _tick;                                              // 秒针的指向
    EventLoop *_event_loop;                                      // 用于监控timerfd的event_loop
    int _timerfd;                                                // 未来实际执行时的timerfd
    std::unique_ptr<Channel> _timerfd_channel;                   // 根据timerfd构建的Channel，用于管理timerfd的事件（读回调，把数据全清空）
};

// EventLoop与Channel关联
class EventLoop
{
    using event_func_t = std::function<void()>;

private:
    // 判断是否在同一个线程
    bool IsSameThread()
    {
        return _thread_id == std::this_thread::get_id();
    }

    // 唤醒用于通知任务队列有数据的eventfd（向里面写入个数）
    void WakeupEventfd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if (ret < 0)
        {
            ERROR_LOG("WakeupEventfd Error!");
            abort();
        }
    }

    // 处理eventfd的读事件（把任务读出来）
    void EventfdHandler()
    {
        uint64_t rval = 0;
        // 这里已经设置了非阻塞
        int ret = read(_event_fd, &rval, sizeof(rval));
        if (ret < 0)
        {
            // 非异常情况
            if (errno == EINTR || errno == EAGAIN)
                return;
            ERROR_LOG("EventfdHandler Error!");
            abort();
        }
    }

public:
    EventLoop() : _epoller(std::make_unique<Epoller>()), _thread_id(std::this_thread::get_id()), _time_wheel(this)
    {
        // 初始化_event_fd
        _event_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
        if (_event_fd < 0)
        {
            ERROR_LOG("create eventfd error!");
            // 理论不应出现这个错误，出现了终止进程
            abort();
        }

        // 初始化_event_channel
        _event_channel = std::make_unique<Channel>(_event_fd, this);
        // 设置eventfd的读回调
        _event_channel->SetReadCallback(std::bind(&EventLoop::EventfdHandler, this));
        // 设置对eventfd读事件的关心
        _event_channel->SetReadCare();
    }
    void Update(Channel *channel)
    {
        _epoller->Update(channel);
    }
    void Delete(Channel *channel)
    {
        _epoller->Delete(channel);
    }
    // 确保其它模块在本EventLoop中执行任务（直接执行 or 放入队列）
    void RunTaskInLoop(const event_func_t &task_func)
    {
        // DEBUG_LOG("IsSameThread:%d",IsSameThread());
        if (IsSameThread())
            // 直接处理
            task_func();
        else
            PushTaskInQueuqe(task_func);
    }
    // 将操作放入任务池
    void PushTaskInQueuqe(const event_func_t &cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _task_queue.push_back(cb);
        }
        // 唤醒有可能因为没有事件就绪，导致的epoll阻塞；
        // 其实就是给eventfd写入一个数据，eventfd就会触发可读事件
        WakeupEventfd();
    }

    // 执行任务队列中的任务
    void RunTaskQueue()
    {
        std::vector<event_func_t> task_queue;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _task_queue.swap(task_queue); // 取出任务
        }
        // 执行任务
        for (auto &task : task_queue)
        {
            task();
        }
    }

    void EnsureSameThread()
    {
        assert(IsSameThread());
        return;
    }

    // 开始监控-处理事件
    void Start()
    {
        while (true)
        {
            std::vector<Channel *> ready_channel;
            // 监控事件
            _epoller->WaitRun(ready_channel);
            // 处理事件
            for (auto &channel : ready_channel)
            {
                channel->EventHandler();
            }
            // 执行任务队列
            RunTaskQueue();
        }
    }

    void AddTimeTask(uint64_t id, uint32_t timeout, task_fun_t task_fun)
    {
        // DEBUG_LOG("AddTimeTask");
        _time_wheel.AddTaskInLoop(id, timeout, task_fun);
    }
    void RefreshTimeTask(uint64_t id)
    {
        _time_wheel.RefreshTaskInLoop(id);
    }
    // 暂时用不上
    void DeleteTimeTask(uint64_t id)
    {
        _time_wheel.DeleteTaskInLoop(id);
    }
    void CancelTimeTask(uint64_t id)
    {
        _time_wheel.CancelTimeTaskInLoop(id);
    }
    bool TimeTaskExist(uint64_t id)
    {
        return _time_wheel.TimeTaskExist(id);
    }

private:
    std::unique_ptr<Epoller> _epoller;       // 管理的Epoll模型
    std::vector<event_func_t> _task_queue;   // 存储的任务队列
    int _event_fd;                           // 用于记录未来通知监控事件就绪的event_fd（防止任务队列有任务，却阻塞在监控事件处）
    std::unique_ptr<Channel> _event_channel; // event_fd构建的Channel，用于放入Epoll模型监控event_fd
    std::thread::id _thread_id;              // 记录所在线程的id
    std::mutex _mutex;                       // 保证多线程访问任务队列的安全
    TimeWheel _time_wheel;                   // 时间轮定时器模块
};

// 在某一线程下实例化Loop
class LoopThread
{
private:
    // 线程的入口函数，在这里实例化EventLoop
    void EntryRoutine()
    {
        EventLoop event_loop;   //创建该线程的event_loop
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _event_loop_ptr = &event_loop;
            _cond.notify_all();
        }
        event_loop.Start(); // 在这个线程内死循环的监控
    }

public:
    LoopThread() : _thread(std::thread(&LoopThread::EntryRoutine, this)), _event_loop_ptr(nullptr)
    {
    }
    EventLoop *GetEventLoop()
    {
        EventLoop *loop = nullptr;
        {
            // 对_event_loop_ptr的操作需要加锁
            // 不满足条件会去条件变量下等待
            std::unique_lock<std::mutex> _lock(_mutex);
            _cond.wait(_lock, [&]() -> bool
                       { return _event_loop_ptr != nullptr; });
            loop = _event_loop_ptr;
        }
        // DEBUG_LOG("loop:%p", loop);
        return loop;
    }

private:
    std::thread _thread;           // EventLoop所在的线程
    EventLoop *_event_loop_ptr;    // 存储所创建的EventLoop的指针，用于返回给上层
    std::mutex _mutex;             // 不同线程访问_event_loop_ptr需要加锁
    std::condition_variable _cond; // 如果此时还没有创建_event_loop，则在条件变量下等待
};

// 用于管理 LoopThread
class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop *master_loop)
        : _next_index(0), _thread_count(0), _master_loop(master_loop)
    {
    }
    // 创建从线程
    void CreateThreadLoops(size_t count)
    {
        _thread_count = count;
        if (_thread_count > 0)
        {
            _loop_threads.resize(_thread_count);
            _slave_loops.resize(_thread_count);
            for (int i = 0; i < _thread_count; i++)
            {
                _loop_threads[i] = new LoopThread(); // 此时从线程已经启动
                _slave_loops[i] = _loop_threads[i]->GetEventLoop();
            }
        }
    }

    EventLoop *SelectOneLoop()
    {
        if (_thread_count == 0)
            return _master_loop;
        _next_index = (_next_index + 1) % _thread_count;
        return _slave_loops[_next_index];
    }

private:
    size_t _next_index;                      // 选取的下一个从Loop的index
    size_t _thread_count;                    // 线程的数量（0个则直接返回master）
    EventLoop *_master_loop;                 // 记录主EventLoop
    std::vector<LoopThread *> _loop_threads; // 管理的所有LoopThread
    std::vector<EventLoop *> _slave_loops;   // 管理的LoopThread中所有的Loop
};

// 外部会根据文件描述符构建Connection对象
// 设置是否超时释放
// 调用Read（从缓冲区读取数据）
// 调用Send（向缓冲区写数据）
//  协议切换（要求立即执行）
class Connection;
using ConnectionPtr = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{

    // 由上层传入的各种阶段的回调函数：连接刚建立，任意事件，连接释放，收到数据后（业务处理）
    using establish_func_t = std::function<void(ConnectionPtr)>;
    using any_event_func_t = std::function<void(ConnectionPtr)>;
    using release_func_t = std::function<void(ConnectionPtr)>;      // 用于用户提供的释放前操作
    using serv_release_func_t = std::function<void(ConnectionPtr)>; // 服务器提供的释放（从服务器维护的结构中删除自身）
    using msg_handler_func_t = std::function<void(ConnectionPtr, Buffer *)>;
    typedef enum
    {
        DISCONNECTED /*未建立连接*/,
        CONNECTING /*连接建立中*/,
        CONNECTED /*连接已建立*/,
        WILL_CONNECTED /*连接待关闭*/
    } conn_status_t;

private:
    // 设置给Channel的事件处理函数
    //  读事件触发  1.读取数据到接收缓冲区 2.业务处理（可以放到里面直接处理，也可以让外部自己拿数据处理）
    void ReadEventHandler() // 把数据从Socket当中读到缓冲区
    {
        // 1.读取数据到接收缓冲区
        char buff[65536] = {0};
        // 创建Socket的时候设置了非阻塞
        ssize_t ret = _socket.Recv(buff, sizeof(buff) - 1);
        if (ret < 0)
        {
            // 读取出错了，把缓冲区数据处理完再关闭
            DisConnect();
            return;
        }
        // 对于自己实现的Recv接口，返回值如果等于0，不是说对端关闭连接，而是没数据可读了，所以正常处理即可
        _inbuffer.Write(buff, ret);
        // 2.业务处理（可选）
        if (_inbuffer.ReadSize() > 0)
        {
            // DEBUG_LOG("准备进行业务处理");
            _msg_handler_func(shared_from_this(), &_inbuffer);
        }
    }
    void WriteEventHandler()
    {
        // 把发送缓冲区的数据发出去
        int ret = _socket.Send(_outbuffer.ReadPtr(), _outbuffer.ReadSize());
        if (ret < 0)
        {
            // 发送出错了，检查一下接收缓冲区还有没有数据，处理一下直接关闭连接
            if (_inbuffer.ReadSize() > 0)
            {
                _msg_handler_func(shared_from_this(), &_inbuffer);
            }
            // 关闭连接
            ReleaseConnOnLast();
        }
        else //>0
        {
            // DEBUG_LOG("send ret = %d", ret);
            // 移动发送缓冲区读偏移
            _outbuffer.UpdateReadOffset(ret);
            if (_outbuffer.ReadSize() == 0)
            {
                // 本次发送完毕，关闭对写事件的关心，也可能socket缓冲区真的满了，那就继续关心写事件
                _channel.CancelWirteCare();
            }

            // 用户可能主动断开了连接，处理一下
            if (_conn_status == WILL_CONNECTED)
                ReleaseConnOnLast();
        }
    }

    // 挂断事件的触发一般都是对端关闭了连接，就不必再发数据了
    void CloseEventHandler()
    {
        // 连接挂断之前把待处理数据处理一下
        // 修改状态
        _conn_status == WILL_CONNECTED;
        // 处理完数据后再关闭
        if (_inbuffer.ReadSize() > 0)
            _msg_handler_func(shared_from_this(), &_inbuffer);
        ReleaseConnOnLast();
    }
    void ErrorEventHandler()
    {
        // 同样挂断即可
        CloseEventHandler();
    }
    // 主要处理两点任务：1、根据是否设置超时断联 刷新任务 2、调用上层提供的任意事件发生接口
    void AnyEventHandler()
    {
        // 如果设置了，刷新一下
        if (_enable_timeout_release)
        {
            DEBUG_LOG("刷新了一下%d的活跃度", _socket.Fd());
            _event_loop->RefreshTimeTask(_connect_id);
        }
        // 调用上层提供的接口
        if (_any_event_func)
            _any_event_func(shared_from_this());
    }

    // 建立连接后调用的接口，一方面要进一步初始化连接，另一方面要调用用户提供的回调
    void Establish()
    {
        assert(_conn_status == CONNECTING);
        // 连接初始化这里会设置读事件的关心
        _channel.SetReadCare();
        // 调用用户提供的接口
        if (_establish_func)
        {
            // ERROR_LOG("准备调用Establish");
            _establish_func(shared_from_this());
        }

        _conn_status = CONNECTED;
    }

    // 真正释放连接的接口
    void ReleaseConn()
    {

        DEBUG_LOG("准备释放%ld的连接：", _connect_id);
        if (_is_release == true)
            // 修改状态
            _conn_status == DISCONNECTED;
        // 移除监控
        _channel.DeleteFd();
        // 关闭文件描述符
        _socket.Close();
        // 如果有，则取消超时释放任务
        if (_event_loop->TimeTaskExist(_connect_id))
            CancelTimeoutRelease();
        // 先调用用户的回调
        if (_release_func)
            _release_func(shared_from_this());
        // 再调用服务器的回调-这个回调是TcpServer设置的，用于把存储的shared_ptr删除
        if (_serv_release_func)
            _serv_release_func(shared_from_this());
    }
    // 启动非活跃连接的释放
    void EnableTimeoutRelease(uint32_t timeout)
    {
        _enable_timeout_release = true;
        if (_event_loop->TimeTaskExist(_connect_id))
        {
            // 存在则刷新一下就行
            _event_loop->RefreshTimeTask(_connect_id);
        }
        else
        {
            // 不存在则添加-绑定释放连接的函数
            _event_loop->AddTimeTask(_connect_id, timeout, std::bind(&Connection::ReleaseConnOnLast, this));
        }
    }
    // 取消非活跃连接的释放
    void CancelTimeoutRelease()
    {
        // DEBUG_LOG("CancelTimeoutRelease");
        _enable_timeout_release = false;
        // 存在则取消
        if (_event_loop->TimeTaskExist(_connect_id))
        {
            _event_loop->CancelTimeTask(_connect_id);
        }
    }
    // 供用户使用的关闭连接接口（会处理输入输出缓冲区再关闭）
    void DisConnect()
    {
        // 修改状态，等到写事件处理完就会判断是否设置关闭连接了
        _conn_status = WILL_CONNECTED;
        // 处理输入缓冲区
        if (_inbuffer.ReadSize() > 0)
            _msg_handler_func(shared_from_this(), &_inbuffer);
        // 处理发送缓冲区
        if (_outbuffer.ReadSize() > 0)
        {
            if (!_channel.IsSetWriteCare())
                _channel.SetWriteCare(); // 设置写关心
        }
        // 如果发生缓冲区没数据了，直接关闭
        if (_outbuffer.ReadSize() == 0)
        {
            ReleaseConnOnLast();
        }
    }
    // 切换协议以及上下文
    void ProtocalSwitch(Any &protocal_contect,
                        const establish_func_t &establish_func,
                        const any_event_func_t &any_event_func,
                        const release_func_t &release_func,
                        const msg_handler_func_t &msg_handler_func)
    {
        _protocol_context = protocal_contect;
        _establish_func = establish_func;
        _any_event_func = any_event_func;
        _release_func = release_func;
        _msg_handler_func = msg_handler_func;
    }

    // 给用户提供的发送接口，不是真正的发送，而是把数据发到了发送缓冲区，并设置写关心
    void Send(Buffer &temp_buffer)
    {
        if (_conn_status == DISCONNECTED)
            return;
        _outbuffer.Write(temp_buffer.ReadPtr(), temp_buffer.ReadSize());
        //_outbuffer.Write(data, len);
        // DEBUG_LOG("len = %d",len);
        //  设置写关心
        if (!_channel.IsSetWriteCare())
            _channel.SetWriteCare();
    }

public:
    Connection(uint64_t id, int sockfd, EventLoop *event_loop)
        : _connect_id(id), _enable_timeout_release(false), _sockfd(sockfd), _socket(sockfd), _conn_status(CONNECTING), _event_loop(event_loop), _channel(sockfd, event_loop)
    {
        // 设置读事件关心不应该在构造函数，因为此时可能还没有设置各种回调，尤其是还没添加定时任务
        _channel.SetReadCallback(std::bind(&Connection::ReadEventHandler, this));
        _channel.SetWriteCallback(std::bind(&Connection::WriteEventHandler, this));
        _channel.SetCloseCallback(std::bind(&Connection::CloseEventHandler, this));
        _channel.SetAnyCallback(std::bind(&Connection::AnyEventHandler, this));
        _channel.SetErrorCallback(std::bind(&Connection::ErrorEventHandler, this));
    }
    // 设置各种回调函数（设置之前还未开启监控）
    void SetEstablishCallback(const establish_func_t &establish_func)
    {
        _establish_func = establish_func;
    }
    void SetAnyEventCallback(const any_event_func_t &any_event_func)
    {
        _any_event_func = any_event_func;
    }
    void SetReleaseCallback(const release_func_t &release_func)
    {
        _release_func = release_func;
    }
    void SetServReleaseCallback(const serv_release_func_t &serv_release_func)
    {
        _serv_release_func = serv_release_func;
    }
    void SetMsgHandlerCallback(const msg_handler_func_t &msg_handler_func)
    {
        _msg_handler_func = msg_handler_func;
    }

    // 实际关闭时的接口，为了确保释放连接是在本轮事件处理完后进行，再封装一层，当超时事件就绪，把真正释放的操作放入任务队列中
    // 防止后面的事件还没处理，就已经被释放了
    void ReleaseConnOnLast()
    {
        DEBUG_LOG("将%p，connid = %ld的释放放入本轮最后执行", shared_from_this().get(), _connect_id);
        if (_is_release == true)
        {
            DEBUG_LOG("检测到%p，connid = %ld释放操作被放入过，本次不放入", shared_from_this().get(), _connect_id);
            return;
        }
        _is_release = true;
        _event_loop->PushTaskInQueuqe(std::bind(&Connection::ReleaseConn, this));
    }
    // 提供给用户的关闭连接接口（实际上不是直接关闭连接，会把输入输出缓冲区的数据处理一下再关闭）
    void DisConnectInLoop()
    {
        _event_loop->RunTaskInLoop(std::bind(&Connection::DisConnect, this));
    }
    // 给用户提供的发送接口，不是真正的发送，而是把数据发到了发送缓冲区，并设置写关心
    // 未来用户的业务处理完，就会调用这个接口向缓冲区内写入数据
    void SendInLoop(const void *data, size_t len)
    {
        // 注意外界传入的data，可能是一个临时的空间，我们把发送任务压入队列，等到执行的时候可能就被释放了
        // 所以我们转存一下
        Buffer temp_buffer;
        temp_buffer.Write(data, len);
        _event_loop->RunTaskInLoop(std::bind(&Connection::Send, this, std::move(temp_buffer)));
    }
    /*这里并没有给用户提供接收接口，而是直接选择了默认接收完后就立即处理，实际上可以选择给用户提供接收接口
    让用户来从_inbuffer中读数据，自己调用函数处理*/
    void EstablishInLoop()
    {
        _event_loop->RunTaskInLoop(std::bind(&Connection::Establish, this));
    }
    void EnableTimeoutReleaseInLoop(uint32_t timeout)
    {
        _event_loop->RunTaskInLoop(std::bind(&Connection::EnableTimeoutRelease, this, timeout));
    }
    void CancelTimeoutReleaseInLoop()
    {
        _event_loop->RunTaskInLoop(std::bind(&Connection::CancelTimeoutRelease, this));
    }

    // 这个函数要保证在EventLoop线程内立即被处理，否则可能因为协议切换不及时而造成使用旧的协议
    void ProtocalSwitchInLoop(Any &protocal_contect,
                              const establish_func_t &establish_func,
                              const any_event_func_t &any_event_func,
                              const release_func_t &release_func,
                              const msg_handler_func_t &msg_handler_func)
    {
        _event_loop->EnsureSameThread(); // 一定在EventLoop线程内
        _event_loop->RunTaskInLoop(std::bind(&Connection::ProtocalSwitch, this, protocal_contect, establish_func, any_event_func, release_func, msg_handler_func));
    }
    uint64_t Id()
    {
        return _connect_id;
    }

    // 设置上下文，连接建立成功后设置，都是被间接调用的函数，不用InLoop
    void SetContent(const Any &content)
    {
        _protocol_context = content;
    }

    Any *GetContent()
    {
        return &_protocol_context;
    }

private:
    uint64_t _connect_id;         // 标识连接的唯一性
    bool _enable_timeout_release; // 是否启用超时断联
    int _sockfd;                  // 关联的文件描述符
    conn_status_t _conn_status;   // 当前连接的状态
    TcpSocket _socket;            // 管理的TcpSocket
    EventLoop *_event_loop;       // 关联的事件监控
    Channel _channel;             // 管理的事件
    Buffer _inbuffer;             // 输入缓冲区
    Buffer _outbuffer;            // 输出缓冲区
    Any _protocol_context;        // 保存的协议上下文
    establish_func_t _establish_func;
    any_event_func_t _any_event_func;
    release_func_t _release_func;
    serv_release_func_t _serv_release_func;
    msg_handler_func_t _msg_handler_func;
    bool _is_release = false; // 记录该连接是否被释放过，防止重复释放
};

// 对监听套接字进行管理的模块
class Acceptor
{
    using accept_func_t = std::function<void(int)>; // 对新连接处理的函数，由上层提供
private:
    // 读事件触发的回调：1、接收新连接 2、调用上层传入的回调accept_fun处理新连接
    void ReadHandler()
    {
        // 1. 默认行为：接收新连接
        // DEBUG_LOG("尝试接收新连接");
        int newfd = _socket.Accept();
        // DEBUG_LOG("newfd = %d", newfd);
        if (newfd < 0)
            return;
        // 2. 调用上层传过来的回调，对newfd处理
        if (_accept_func)
            _accept_func(newfd);
    }
    int CreateTcpSocket(uint16_t port)
    {
        int ret = _socket.ServerCreate(port, SOCKFD_NONBLOCK);
        if (ret < 0)
        {
            ERROR_LOG("CreateTcpSocket Error!");
            abort();
        }
        return ret; // 得到文件描述符用于构造Socket，构造Channel
    }

public:
    // 由于没实现Channel的默认构造，所以应该在初始化列表构造Channel，但是构造Channel需要文件描述符，所以我们需要使用自己写的函数先构造一个Socket
    Acceptor(EventLoop *event_loop, uint16_t port)
        : _event_loop(event_loop), _socket(CreateTcpSocket(port)), _channel(_socket.Fd(), event_loop)
    {
        DEBUG_LOG("建立监听套接字：fd = %d", _socket.Fd());
        // 为监听套接字设置读事件触发回调
        _channel.SetReadCallback(std::bind(&Acceptor::ReadHandler, this));
        // 注意启动读事件关心（监听），不能在构造函数内，否则可能启动监控立即有事件，处理的时候回调函数还未设置
    }
    void ReadCareListen()
    {
        _channel.SetReadCare(); // 或者调用EventLoop的接口也能启动
    }
    void SetAcceptCallback(const accept_func_t &accept_func)
    {
        // DEBUG_LOG("SetAcceptCallback");
        _accept_func = accept_func;
    }

private:
    TcpSocket _socket;          // 管理的套接字
    Channel _channel;           // 管理的事件
    EventLoop *_event_loop;     // 用于监控事件的loop
    accept_func_t _accept_func; // 上层对新连接处理的回调
};

class TcpServer
{
    // 用于传给Connection对象的回调函数
    using establish_func_t = std::function<void(ConnectionPtr)>;
    using any_event_func_t = std::function<void(ConnectionPtr)>;
    using release_func_t = std::function<void(ConnectionPtr)>;      // 用于用户提供的释放前操作
    using serv_release_func_t = std::function<void(ConnectionPtr)>; // 服务器提供的释放（从服务器维护的结构中删除自身）
    using msg_handler_func_t = std::function<void(ConnectionPtr, Buffer *)>;

private:
    void CreateNewConn(int new_sockfd)
    {
        _key_id++;
        // 为新连接创建Connection，并绑定至指定线程
        DEBUG_LOG("主线程收到了新的连接:fd = %d", new_sockfd);
        ConnectionPtr new_sock_connection(new Connection(_key_id, new_sockfd, _loop_thread_pool.SelectOneLoop()));
        // 为通信套接字设置回调(由用户传给TcpServer)
        new_sock_connection->SetMsgHandlerCallback(_msg_handler_func);
        new_sock_connection->SetReleaseCallback(_release_func);
        new_sock_connection->SetAnyEventCallback(_any_event_func);
        new_sock_connection->SetEstablishCallback(_establish_func);
        new_sock_connection->SetServReleaseCallback(std::bind(&TcpServer::DeleteConnInLoop, this, std::placeholders::_1));
        if (_enable_timeout_release)
            new_sock_connection->EnableTimeoutReleaseInLoop(_release_timeout); // 开启定时销毁任务
        DEBUG_LOG("主线程开启对:fd = %d的监控", new_sockfd);
        // 初始化连接，设置读事件监控，顺便调用连接建立成功的方法
        new_sock_connection->EstablishInLoop();
        // 创建好之后，插入进去
        _connections.insert(std::make_pair(_key_id, new_sock_connection));
    }
    void DeleteConn(ConnectionPtr conn_ptr)
    {
        auto it = _connections.find(conn_ptr->Id());
        if (it != _connections.end())
        {
            _connections.erase(it);
        }
    }

public:
    TcpServer(uint16_t port, size_t slave_cnt)
        : _key_id(0), _port(port), _slave_cnt(slave_cnt), _acceptor(&_master_loop, port), _enable_timeout_release(false), _loop_thread_pool(&_master_loop)
    {
        // 给Acceptor绑定回调
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::CreateNewConnInLoop, this, std::placeholders::_1));
        // 开启读事件关心
        _acceptor.ReadCareListen();
        // 创建并启动从属线程
        _loop_thread_pool.CreateThreadLoops(slave_cnt);
    }
    // 开启超时释放任务
    void EnableTimeoutRelease(uint32_t timeout)
    {
        _enable_timeout_release = true;
        _release_timeout = timeout;
    }
    // 用于接收连接后构造新的Connection，此回调传给Acceptor
    void CreateNewConnInLoop(int newfd)
    {
        _master_loop.RunTaskInLoop(std::bind(&TcpServer::CreateNewConn, this, newfd));
    }
    // 未来从服务器中删除存储的Connection的shared_ptr，要确保它在master_loop所在线程执行，避免出现线程安全问题
    void DeleteConnInLoop(ConnectionPtr conn_ptr)
    {
        _master_loop.RunTaskInLoop(std::bind(&TcpServer::DeleteConn, this, conn_ptr));
    }

    // 用户给TcpServer TcpServer给Connection设置的各自回调
    void SetEstablishCallback(const establish_func_t &establish_func)
    {
        _establish_func = establish_func;
    }
    void SetAnyEventCallback(const any_event_func_t &any_event_func)
    {
        _any_event_func = any_event_func;
    }
    void SetReleaseCallback(const release_func_t &release_func)
    {
        _release_func = release_func;
    }
    void SetMsgHandlerCallback(const msg_handler_func_t &msg_handler_func)
    {
        _msg_handler_func = msg_handler_func;
    }

    void Start()
    {
        _master_loop.Start();
    }

private:
    uint64_t _key_id;       // 标识连接和超时任务唯一性的id，各自保持独立性，所以用同一个即可
    EventLoop _master_loop; // 主Reactor的EventLoop
    uint16_t _port;         // 记录服务器绑定的端口
    size_t _slave_cnt;      // 从属线程的数量
    std::unordered_map<uint64_t, ConnectionPtr> _connections;
    Acceptor _acceptor;               // 用于管理监听套接字的对象
    LoopThreadPool _loop_thread_pool; // 从属线程池，管理多个SlaveLoop
    bool _enable_timeout_release;     // 是否设置超时任务
    uint32_t _release_timeout;        // 设置超时释放的时间
    // 多个上层用户设置传给Connection对象的回调函数
    establish_func_t _establish_func;
    any_event_func_t _any_event_func;
    release_func_t _release_func;
    serv_release_func_t _serv_release_func;
    msg_handler_func_t _msg_handler_func;
};

// Channel操作EventLoop的成员函数
void Channel::UpdateToEpoller()
{
    _event_loop->Update(this);
}
void Channel::DeleteFromEpoller()
{
    _event_loop->Delete(this);
}

void TimeWheel::DeleteTaskInLoop(uint64_t id)
{
    _event_loop->RunTaskInLoop(std::bind(&TimeWheel::DeleteTimetask, this, id));
}
void TimeWheel::AddTaskInLoop(uint64_t id, uint32_t timeout, task_fun_t task_fun)
{
    // DEBUG_LOG("AddTaskInLoop");
    _event_loop->RunTaskInLoop(std::bind(&TimeWheel::AddTimeTask, this, id, timeout, task_fun));
}
void TimeWheel::RefreshTaskInLoop(uint64_t id)
{
    _event_loop->RunTaskInLoop(std::bind(&TimeWheel::RefreshTimeTask, this, id));
}
void TimeWheel::CancelTimeTaskInLoop(uint64_t id)
{
    _event_loop->RunTaskInLoop(std::bind(&TimeWheel::CancelTimeTask, this, id));
}

class SigPipeIgn
{
public:
    SigPipeIgn()
    {
        DEBUG_LOG("SigPipeIgn Init");
        signal(SIGPIPE, SIG_IGN);
    }
};

static SigPipeIgn sigpipe_ign;